package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/9 16:12
 */
public class Flink_SQL_Datagen2Kafka {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        STRING,\n" +
                "    ts   TIMESTAMP(3),\n" +
                "    vc INT\n" +
//                "    price        DECIMAL(32,2),\n" +
//                "    buyer        ROW<first_name STRING, last_name STRING>,\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'rows-per-second' = '1', \n" +
                "  'fields.id.length' = '5', \n" +
                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");


//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");
//


        tableEnv.executeSql("create table sinkTable (\n" +
                "   id string,\n" +
                "   ts timestamp(3),\n" +
                "   vc INT\n" +
                ") with ( \n"
                + "'connector' = 'kafka',"
                + "'topic' = 'drity_data_sink',"
                + "'properties.bootstrap.servers' = '192.168.1.2:9092',"
                + "'format' = 'json'"
                + ")");

        tableEnv.executeSql("insert into sinkTable select * from sourceTable");

    }

}
